/** 
* @version:1.0.1
* @Description: （对类进行功能描述）
* @author: yangdechao
* @date: datedate 2021年11月15日 上午10:22:04
*/
package cn.com.guage.flink.source;

import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.PreparedStatement;
import java.sql.ResultSet;

import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.source.RichSourceFunction;

import cn.com.guage.flink.domain.User;

/** 
* @version:1.0.1
* @Description: 读取mysql source
* @author: yangdechao
* @date: datedate 2021年11月15日 上午10:22:04
*/
public class SourceFromMySQL extends RichSourceFunction<User> {

	/**
	 * 
	 */
	private static final long serialVersionUID = 733358971334442267L;
	PreparedStatement ps;
    private Connection connection;

    /**
     * open() 方法中建立连接，这样不用每次 invoke 的时候都要建立连接和释放连接。
     *
     * @param parameters
     * @throws Exception
     */
    @Override
    public void open(Configuration parameters) throws Exception {
    
        super.open(parameters);
        connection = getConnection();
        String sql = "select * from t_user;"; // 编写具体逻辑代码
        ps = this.connection.prepareStatement(sql);
    }

    /**
     * 程序执行完毕就可以进行，关闭连接和释放资源的动作了
     *
     * @throws Exception
     */
    @Override
    public void close() throws Exception {
    
        super.close();
        if (connection != null) {
     //关闭连接和释放资源
            connection.close();
        }
        if (ps != null) {
    
            ps.close();
        }
    }

    public void run(SourceContext<User> ctx) throws Exception {
    
        ResultSet resultSet = ps.executeQuery(); // 执行SQL语句返回结果集
        while (resultSet.next()) {
    
            User User = new User(
                    resultSet.getInt("id"),
                    resultSet.getString("userName"), resultSet.getInt("amount"));
            ctx.collect(User);
        }
    }

    public void cancel() {
    
    }

    private static Connection getConnection() {
    
        Connection con = null;
        try {
    
            Class.forName("com.mysql.jdbc.Driver");
            con = DriverManager.getConnection("jdbc:mysql://localhost:3306/test?useUnicode=true&characterEncoding=UTF-8", "root", "123456");
        } catch (Exception e) {
    
            System.out.println("-----------mysql get connection has exception , msg = "+ e.getMessage());
        }
        return con;
    }

}
